package net.qiqbframework.eventhandling;

import lombok.extern.slf4j.Slf4j;
import net.qiqbframework.config.ConfigurationException;
import net.qiqbframework.common.Registration;
import net.qiqbframework.messaging.intercepting.MessageDispatchInterceptor;
import net.qiqbframework.messaging.MessageHandler;
import net.qiqbframework.messaging.monitoring.MessageMonitor;
import net.qiqbframework.messaging.monitoring.MonitorCallback;
import net.qiqbframework.messaging.monitoring.NoOpMessageMonitor;
import net.qiqbframework.messaging.task.CurrentMessageTask;
import net.qiqbframework.messaging.task.MessageTask;
import net.qiqbframework.persisthanding.DefaultEventBusSpanFactory;
import net.qiqbframework.persisthanding.EventBusSpanFactory;
import net.qiqbframework.tracing.NoOpSpanFactory;

import javax.annotation.Nonnull;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Set;
import java.util.concurrent.CopyOnWriteArraySet;
import java.util.function.BiFunction;
import java.util.function.Consumer;

import static net.qiqbframework.common.BuilderUtils.assertNonNull;

@Slf4j
public class SimpleEventBus implements EventBus {

    private final MessageMonitor<? super EventMessage<?>> messageMonitor;

    private final String eventsKey = this + "_EVENTS";
    private final Set<Consumer<EventMessage<?>>> eventProcessors = new CopyOnWriteArraySet<>();
    private final Set<MessageDispatchInterceptor<? super EventMessage<?>>> dispatchInterceptors = new CopyOnWriteArraySet<>();
    private final EventBusSpanFactory spanFactory;


    protected SimpleEventBus(Builder builder) {
        builder.validate();
        this.messageMonitor = builder.messageMonitor;
        this.spanFactory = builder.spanFactory;
    }

    @Override
    public Registration subscribe(@Nonnull Consumer< EventMessage<?>> eventProcessor) {
        if (this.eventProcessors.add(eventProcessor)) {
            if (log.isDebugEnabled()) {
                log.debug("EventProcessor [{}] subscribed successfully", eventProcessor);
            }
        } else {
            log.info("EventProcessor [{}] not added. It was already subscribed", eventProcessor);
        }
        return () -> {
            if (eventProcessors.remove(eventProcessor)) {
                if (log.isDebugEnabled()) {
                    log.debug("EventListener {} unsubscribed successfully", eventProcessor);
                }
                return true;
            } else {
                log.info("EventListener {} not removed. It was already unsubscribed", eventProcessor);
                return false;
            }
        };
    }

    @Override
    public Registration registerDispatchInterceptor(@Nonnull MessageDispatchInterceptor<? super EventMessage<?>> dispatchInterceptor) {
        dispatchInterceptors.add(dispatchInterceptor);
        return () -> dispatchInterceptors.remove(dispatchInterceptor);
    }

    @Override
    public void publish(@Nonnull EventMessage<?> event) {

        final EventMessage<?> eventMessage = spanFactory.createPublishEventSpan(event)
                .runSupplier(() -> spanFactory.propagateContext(event));
        MonitorCallback ingested = messageMonitor.onMessageIngested(eventMessage);
        spanFactory.createCommitEventsSpan().run(() -> {
            try {
                eventProcessors.forEach(eventProcessor -> eventProcessor.accept(event));
                ingested.reportSuccess();
            } catch (Exception e) {
                ingested.reportFailure(e);
                throw e;
            }
        });
    }

    @Override
    public Registration subscribe(@Nonnull String eventName, @Nonnull MessageHandler<? super EventMessage<?>> handler) {
        return null;
    }


    protected List<EventMessage<?>> queuedMessages() {
        if (!CurrentMessageTask.isStarted()) {
            return Collections.emptyList();
        }
        List<EventMessage<?>> messages = new ArrayList<>();
        addStagedMessages(CurrentMessageTask.get(), messages);
        return messages;
    }

    private void addStagedMessages(MessageTask<?> unitOfWork, List<EventMessage<?>> messages) {
        unitOfWork.parent().ifPresent(parent -> addStagedMessages(parent, messages));
        if (unitOfWork.isRolledBack()) {
            // staged messages are irrelevant if the UoW has been rolled back
            return;
        }
        List<EventMessage<?>> stagedEvents = unitOfWork.getOrDefaultResource(eventsKey, Collections.emptyList());
        for (EventMessage<?> stagedEvent : stagedEvents) {
            if (!messages.contains(stagedEvent)) {
                messages.add(stagedEvent);
            }
        }
    }

    protected List<? extends EventMessage<?>> intercept(List<? extends EventMessage<?>> events) {
        List<EventMessage<?>> preprocessedEvents = new ArrayList<>(events);
        for (MessageDispatchInterceptor<? super EventMessage<?>> preprocessor : dispatchInterceptors) {
            BiFunction<Integer, ? super EventMessage<?>, ? super EventMessage<?>> function =
                    preprocessor.handle(preprocessedEvents);
            for (int i = 0; i < preprocessedEvents.size(); i++) {
                preprocessedEvents.set(i, (EventMessage<?>) function.apply(i, preprocessedEvents.get(i)));
            }
        }
        return preprocessedEvents;
    }

    private void doWithEvents(Consumer<List<? extends EventMessage<?>>> eventsConsumer,
                              List<? extends EventMessage<?>> events) {
        eventsConsumer.accept(events);
    }


    public static class Builder {

        private MessageMonitor<? super EventMessage<?>> messageMonitor = NoOpMessageMonitor.INSTANCE;
        private EventBusSpanFactory spanFactory = DefaultEventBusSpanFactory
                .builder().spanFactory(NoOpSpanFactory.INSTANCE).build();

        public Builder messageMonitor(@Nonnull MessageMonitor<? super EventMessage<?>> messageMonitor) {
            assertNonNull(messageMonitor, "MessageMonitor may not be null");
            this.messageMonitor = messageMonitor;
            return this;
        }


        public Builder spanFactory(@Nonnull EventBusSpanFactory spanFactory) {
            assertNonNull(spanFactory, "SpanFactory may not be null");
            this.spanFactory = spanFactory;
            return this;
        }

        protected void validate() throws ConfigurationException {
            // Method kept for overriding
        }

        public SimpleEventBus build() {
            return new SimpleEventBus(this);
        }
    }
}
